package com.example.startup;


import com.example.config.SysMQConfig;
import com.example.constant.MQConstant;
import com.example.dto.MQListenerDto;
import com.example.listener.SysMQMessageListener;
import com.example.listener.handler.SysMQMessageListenerHandler;
import com.example.message.MqMessageReturn;
import com.example.service.SysServerService;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import lombok.Data;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


/****
 * 基于RabbitMQ的实现
 */

@Data
public class SysRabbitMQStartUpService extends AbstractMQStartUpService {

    Logger logger = LoggerFactory.getLogger(getClass());

    private SysMQConfig sysMQConfig;

    private ConnectionFactory connectionFactory;

    private Connection connection;

    private ExecutorService executorService;

    private SysMQMessageListenerHandler sysMQMessageListenerHandler;

    private SysServerService sysServerService;

    private Map<Integer, Channel> channelMap = new ConcurrentHashMap<>();

    @Override
    public void init(SysMQConfig sysMQConfig, SysMQMessageListenerHandler sysMQMessageListenerHandler, SysServerService sysServerService) {
        try {
            this.sysMQConfig = sysMQConfig;
            this.sysMQMessageListenerHandler = sysMQMessageListenerHandler;
            this.sysServerService = sysServerService;
            this.connectionFactory = new ConnectionFactory();
            this.connectionFactory.setHost(sysMQConfig.getHost());
            this.connectionFactory.setPort(sysMQConfig.getPort());
            this.connectionFactory.setUsername(sysMQConfig.getUsername());
            this.connectionFactory.setPassword(sysMQConfig.getPassword());
            this.connection = this.connectionFactory.newConnection();
        } catch (Exception e) {
            logger.warn("RabbitMQ初始化失败: " + e.getMessage());
            if (this.connection != null) {
                try {
                    this.connection.close();
                } catch (IOException ex) {
                    logger.warn("链接关闭失败: " + ex.getMessage());
                }
            }
        }
    }

    @Override
    public void start() throws Exception {
        this.executorService = Executors.newCachedThreadPool();
        Map<String, MQListenerDto> listenerMap = this.sysMQMessageListenerHandler.getListenerMap();
        for (String name : listenerMap.keySet()) {
            MQListenerDto mqListenerDto = listenerMap.get(name);
            executorService.execute(new MQListenerThread(mqListenerDto));
        }
    }

    @Override
    public void stop() throws Exception {
        for (Channel channel : this.channelMap.values()) {
            if (channel.isOpen()) {
                channel.close();
            }
        }
        if (this.connection != null) {
            try {
                this.connection.close();
            } catch (IOException e) {
                logger.warn("链接关闭失败: " + e.getMessage());
            }
        }
        this.executorService.shutdownNow();
    }

    @Override
    public void send(String type, String name, String message) throws Exception {
        Channel channel = null;
        try {
            channel = this.connection.createChannel();
            channel.basicPublish(name, name, null, message.getBytes(StandardCharsets.UTF_8));
        } finally {
            if (channel != null) {
                channel.close();
            }
        }
    }

    /***
     * 数据拉取线程
     */
    @Data
    class MQListenerThread implements Runnable {

        private MQListenerDto mqListenerDto;

        public MQListenerThread(MQListenerDto mqListenerDto) {
            this.mqListenerDto = mqListenerDto;
        }

        @Override
        public void run() {
            Channel channel = null;
            try {
                channel = connection.createChannel();
                String exchangeName = "exchange." + mqListenerDto.getName();
                if (mqListenerDto.getType().equalsIgnoreCase(MQConstant.TOPIC)) {
                    isInitExchange(channel, BuiltinExchangeType.FANOUT, exchangeName);
                    String queueName = mqListenerDto.getName() + "_" + sysServerService.getCurrentServerKey();
                    channel.queueDeclare(queueName, true, false, true, null);
                    channel.queueBind(queueName, exchangeName, mqListenerDto.getName());
                    channel.basicConsume(queueName, true, new MQDefaultConsumer(mqListenerDto, channel));
                } else {
                    isInitExchange(channel, BuiltinExchangeType.DIRECT, exchangeName);
                    isExistQueue(channel, mqListenerDto.getName(), false, false, false, null);
                    channel.queueBind(mqListenerDto.getName(), exchangeName, mqListenerDto.getName());
                    channel.basicConsume(mqListenerDto.getName(), true, new MQDefaultConsumer(mqListenerDto, channel));
                }
            } catch (Exception e) {
                logger.warn("MQ监听线程启动失败: " + e.getMessage());
            } finally {
                if (channel != null) {
                    channelMap.put(channel.getChannelNumber(), channel);
                }
            }
        }
    }

    //默认处理类
    class MQDefaultConsumer extends DefaultConsumer {

        private MQListenerDto mqListenerDto;

        private Channel channel;

        public MQDefaultConsumer(MQListenerDto mqListenerDto, Channel channel) {
            super(channel);
            this.channel = channel;
            this.mqListenerDto = mqListenerDto;
        }

        @Override
        public Channel getChannel() {
            return channel;
        }

        public void setChannel(Channel channel) {
            this.channel = channel;
        }

        public MQListenerDto getMqListenerDto() {
            return mqListenerDto;
        }

        public void setMqListenerDto(MQListenerDto mqListenerDto) {
            this.mqListenerDto = mqListenerDto;
        }

        /***
         * 消息的处理
         * @throws IOException
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            String message = new String(body, StandardCharsets.UTF_8);
            Object bean = sysMQMessageListenerHandler.getApplicationContext().getBean(mqListenerDto.getBeanName());
            if (bean instanceof SysMQMessageListener) {
                MqMessageReturn messageReturn = null;
                try {
                    messageReturn = ((SysMQMessageListener) bean).onMessage(message);
                } catch (Exception e) {
                    messageReturn = MqMessageReturn.error(e.getMessage());
                } finally {
                    if (messageReturn.isSuccess()) {
                        logger.warn("MQ监听类消费消息成功: " + message);
                    } else {
                        logger.warn("MQ监听类消费消息失败: " + messageReturn.getMsg());
                    }
                }
            }
        }
    }

    /***
     * 判断交换机是否存在, 如果不存在就返回false，如果存在就返回true
     * @param exchangeName
     * @return
     */
    private boolean isInitExchange(Channel channel, BuiltinExchangeType builtinExchangeType, String exchangeName) {
        boolean flag = true;
        try {
            channel.exchangeDeclare(exchangeName, builtinExchangeType);
        } catch (Exception e) {
            logger.warn("exchangeName : " + exchangeName + " 已存在 " + e.getMessage());
            flag = false;
        }
        return flag;
    }

    /***
     * 判断队列是否存在, 如果不存在就返回false，如果存在就返回true
     * @param queueName
     * @return
     */
    private boolean isExistQueue(Channel channel, String queueName, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments) {
        boolean flag = true;
        try {
            channel.queueDeclare(queueName, durable, exclusive, autoDelete, arguments);
        } catch (Exception e) {
            logger.warn("queueName : " + queueName + " 已存在 " + e.getMessage());
            flag = false;
        }
        return flag;
    }
}
